package com.ibm.cps.message;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.codec.binary.Hex;
import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Throwables;
import com.ibm.datamodel.DataSourceTopic;
import com.ibm.datamodel.timeseries.parser.ProtostuffMessageParser;
import com.ibm.factories.MessageParserFactory;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.IPersistency;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.SchemaCache;
import com.ibm.util.exception.CPSException;

public class MessageFactory {
	public static enum MessageType {
		FEEDBACK_INFO, PROCESSORS, DELETE_PROCESSORS, SLIDING_WINDOW, DFT, DATASOURCE_STATE_CHANGE, ADD_RULE_ENGINE, DELETE_RULE_ENGINE, NEW_RULE_ENGINE, SERVER_STATUS, JSON_DATA_MESSAGE, JSON_NEW_DATA_MESSAGE, ADD_SCHEMA, DELETE_SCHEMA, ADD_RULE_TOPIC, DELETE_RULE_TOPIC, ADD_DATA_SOURCE_TOPIC, DELETE_DATA_SOURCE_TOPIC, MESSAGE_COUNT, PLACE_HOLDER
	};

	private static Logger logger = Logger.getLogger(MessageFactory.class);

	public static final int ID_LENGTH = 12;

	public static byte[] generateBinaryDataMessage(String tenantidStr,
			String schemaIdStr, String sourceIdStr, String data) {
		ObjectNode node = JSONUtility.newObjectNode();
		node.put(MessageFields.TENANTID, tenantidStr);
		node.put(MessageFields.SCHEMAID, schemaIdStr);
		node.put(MessageFields.NAME, sourceIdStr);
		node.put("Message", data);
		return node.toString().getBytes();
	}

	static public RawDataMessage produceDataMessage(byte[] data)
			throws CPSException {
		String json = new String(data);
		ObjectNode node = JSONUtility.fromObjectString(json);
		String tenantid = node.get(MessageFields.TENANTID).asText();
		String schemaId = node.get(MessageFields.SCHEMAID).asText();
		String sourceid = node.get(MessageFields.NAME).asText();
		byte[] msgData = node.get("Message").asText().getBytes();
		if (schemaCache == null) {
			schemaCache = SchemaCache.getInstace();
		}
		
		ArrayList<String> schematype = new ArrayList<String>();
		
		ITimeSeriesSchema schema = schemaCache.getSchema(tenantid, schemaId,schematype);
		if (schema == null) {
			return null;
		}
		
		String type = schematype.get(0);
		IMessageParser parser = MessageParserFactory.createMessageParser(type, schema);
		Comparable tsKey = parser.getTimeSeriesKey(msgData);
		
		RawDataMessage message = new RawDataMessage(tenantid, schemaId,
				sourceid, tsKey, msgData);
		return message;
	}

	// public static byte[] generateBinaryDataMessage(String tenantidStr,
	// String schemaIdStr, String sourceIdStr, byte[] data)
	// throws CPSException {
	// byte[] tenantid = null;
	// byte[] schemaId = null;
	// byte[] sourceId = null;
	// try {
	// tenantid = Hex.decodeHex(tenantidStr.toCharArray());
	// } catch (DecoderException e) {
	// throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "Invalid id "
	// + tenantidStr);
	// }
	//
	// try {
	// schemaId = Hex.decodeHex(schemaIdStr.toCharArray());
	// } catch (DecoderException e) {
	// throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "Invalid id "
	// + schemaIdStr);
	// }
	//
	// try {
	// sourceId = Hex.decodeHex(sourceIdStr.toCharArray());
	// } catch (DecoderException e) {
	// throw new CPSException(ErrorCode.PARSE_ERROR_CODE,
	// "Invalid source id " + sourceIdStr);
	// }
	// return generateBinaryDataMessage(tenantid, schemaId, sourceId, data);
	// }

	public static byte[] generateBinaryDataMessage(byte[] tenantid,
			byte[] schemaId, byte[] sourceId, byte[] data) throws CPSException {

		ByteArrayOutputStream dataStream = new ByteArrayOutputStream();
		if (tenantid == null || tenantid.length != ID_LENGTH
				|| schemaId == null || schemaId.length != ID_LENGTH
				|| sourceId == null || sourceId.length != ID_LENGTH) {
			throw new CPSException(ErrorCode.PARSE_ERROR_CODE,
					"icorrect id format ");
		}

		try {
			dataStream.write(tenantid);
			dataStream.write(schemaId);
			dataStream.write(sourceId);
			dataStream.write(data);
			return dataStream.toByteArray();
		} catch (IOException e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.PARSE_ERROR_CODE, e.getMessage());
		}
	}

	static private SchemaCache schemaCache = null;

	// static public RawDataMessage produceDataMessage(byte[] data)
	// throws CPSException {
	// ByteArrayInputStream dataStream = new ByteArrayInputStream(data);
	//
	// try {
	// String tenantid = readID(dataStream);
	// String schemaId = readID(dataStream);
	// String sourceid = readID(dataStream);
	// byte[] msgData = IOUtils.toByteArray(dataStream);
	// if (schemaCache == null) {
	// schemaCache = SchemaCache.getInstace();
	// }
	// ITimeSeriesSchema schema = schemaCache
	// .getSchema(tenantid, schemaId);
	// if (schema == null) {
	// return null;
	// }
	// long tsKey = schema.getTimeSeriesKey(msgData);
	// RawDataMessage message = new RawDataMessage(tenantid, schemaId,
	// sourceid, tsKey, msgData);
	// return message;
	// } catch (IOException e) {
	// logger.error(e.getClass() + " - " + e.getMessage());
	// logger.error(Throwables.getStackTraceAsString(e));
	// throw new CPSException(ErrorCode.PARSE_ERROR_CODE, e.getMessage(),
	// e);
	// }
	// }

	public static ParsedDataMessage produceParsedDataMessage(IMessageParser parser, RawDataMessage message)
			throws CPSException {
		if (message instanceof RawDataMessage) {
			RawDataMessage rawMsg = (RawDataMessage) message;

			ParsedDataMessage parsedMsg = parser.parseDataMessage(rawMsg);
			
			//modify by gxj
//			ParsedDataMessage parsedMsg = schema.parseDataMessage(rawMsg);
			if (parsedMsg.tsKey == null) {
				throw new CPSException(ErrorCode.PARSE_ERROR_CODE,
						"must specify the time series key");
			}
			return parsedMsg;
		}
		if (message == null) {
			throw new CPSException(ErrorCode.NULL, "The message is null");
		}
		throw new CPSException(ErrorCode.NO_IMPLEMENTATION_CODE,
				"no implementation for current data format");
	}

	private static String readID(ByteArrayInputStream dataStream)
			throws IOException {

		byte[] idBytes = new byte[ID_LENGTH];
		dataStream.read(idBytes);
		return Hex.encodeHexString(idBytes);
	}

	@Deprecated
	public static AbstractMessage produceAbstractMessage(String type,
			byte[] message) {
		type = type.toLowerCase();
		if (type.equals(MessageFields.RULE_TYPE)) {
			return new RuleMetaMessage(message);
		} else if (type.equals(MessageFields.SCHEMA_TYPE)) {
			return new SchemaMetaMessage(message);
		} else if (type.equals(MessageFields.DFT_TYPE)) {
			return new DFTMetaMessage(message);
		} else if (type.equals(MessageFields.INPUT_TYPE)) {
			return new DataSourceTopic(message);
		} else {
			return null;
		}
	}

	/**
	 * @param tenantid
	 * @return
	 */
	public static Collection<String> loadExistedMessage(String tenantid)
			throws CPSException {
		IPersistency persistency = MongoPersistency.getInstace();
		Collection<String> metadatas = persistency.getProcessors(tenantid,
				null, null, null);
		if (metadatas == null) {
			return null;
		}
		List<ObjectNode> sorted = MetadataTopologySort
				.getSortedMetadatas(metadatas);
		metadatas.clear();
		for (ObjectNode node : sorted) {
			metadatas.add(node.toString());
		}
		return metadatas;
	}
	public static String loadtenantId (String data) throws Exception
	{
		String reg = "(^"+MessageType.DELETE_PROCESSORS
				+ "|^"+MessageType.ADD_DATA_SOURCE_TOPIC
				+ "|^"+MessageType.DELETE_DATA_SOURCE_TOPIC
				+ "|^"+MessageType.JSON_DATA_MESSAGE
				+ "|^"+MessageType.PROCESSORS
				+ ")([a-zA-Z0-9]{24})(-[0-9])?";
		Pattern p = Pattern.compile(reg);
		Matcher m = p.matcher(data);
		if(m.find()) {
			logger.info("tenantid:"+m.group(2));
			return m.group(2);
		}
		else {
			throw new Exception("fail to catch tenantid from data");
		}
	}
}
